
package net.neptech.listener;

import lombok.extern.slf4j.Slf4j;
import net.neptech.model.appointment.Appointment;
import net.neptech.model.appointment.ClnicStaff;
import net.neptech.model.dto.AppointmentDTO;
import net.neptech.service.IClnicStaffService;
import net.neptech.topics.ClinicStaffTopics;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.protocol.Message;
import org.apache.kafka.common.protocol.types.RawTaggedField;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.retry.annotation.Backoff;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * @Author ruirui
 * @desc
 * @Date 2025/8/15 16:04
 * @Version 1.0
 */
@Service
@Slf4j
public class ClnicStaffListeners {
    @Autowired
    private KafkaTemplate<String ,Object> kafkaTemplate;
    @Autowired
    private IClnicStaffService iClnicStaffService;

    @RetryableTopic(
            attempts = "3",
            backoff = @Backoff(delay = 1000, multiplier = 2.0),
            retryTopicSuffix = "-retry",
            dltTopicSuffix = "-dlt"
    )
    @KafkaListener(topics = ClinicStaffTopics.SLOT_USE_UPDATE)
    public void slotUseUpdate(@Payload AppointmentDTO payload,
                              @Header(KafkaHeaders.RECEIVED_KEY) String key,
                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                              Acknowledgment ack) {
        try {
            // 1. 处理你的业务逻辑
           int updateNum = key.equals("--slot")? 1 : -1 ;

            Integer i = iClnicStaffService.updateClinicStaffSlotById(payload.getClnicId(), updateNum);
            //确认
            if (i==1)
                ack.acknowledge();
        } catch (Exception ex) {
            // 不要手动发送DLT；抛出交给框架产生重试/死信
            throw ex;
        }
    }
    @KafkaListener(topics = ClinicStaffTopics.PUT_SLOT)
    public void addClinic(@Payload ClnicStaff payload,
                          @Header(KafkaHeaders.RECEIVED_KEY) String key,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                          Acknowledgment ack) {
        try {
            // 1. 处理你的业务逻辑

            boolean r = iClnicStaffService.putSlot(payload);
            //确认
            if (r)
                ack.acknowledge();
        } catch (Exception ex) {
            // 不要手动发送DLT；抛出交给框架产生重试/死信
            log.error("添加失败：{} ,clinicStaff: {}",ex.getMessage(),payload);
            throw ex;
        }
    }
}
